package com.tensai.study.java.newconcept.typereactiveprogram.reactor.flux;


import com.tensai.study.java.newconcept.typereactiveprogram.reactor.schedule.ScheduleTest;
import org.junit.Test;
import reactor.core.publisher.Flux;
import reactor.core.scheduler.Schedulers;

/**
 * 切换执行线程
 *
 * @author Tensai
 */
public class PublishOnTest {

    public static int F = 5;

    private static void cOut(String tag, Object i) {
        System.out.println(Thread.currentThread().getName() + "\t" + tag + "\t" + i + "\t" + Thread.currentThread()
                                                                                                   .getName());
    }

    /**
     * publishOn可以在链式调用的过程中，进行publish的切换：
     *
     * @see ScheduleTest#publishOnTest()
     */
    @Test
    public void test() throws InterruptedException {
        new ScheduleTest().publishOnTest();
        // 可以看到,主线程的名字是Thread。Subscriber线程的名字是ThreadA。
        //
        // 那么在publishOn之前，map使用的线程就是ThreadA。
        // 而在publishOn之后，map使用的线程就切换到了parallel-scheduler线程池。
    }

    @Test
    public void test11() {
        Flux
                //创建消费者
                .create(emitter -> {
                            for (int i = 0; i <= F; i++) {
                                if (emitter.isCancelled()) {
                                    return;
                                }
                                cOut(">>>", i);
                                emitter.next(i);
                            }
                        }
                )
                .doOnNext(s -> cOut("<<<", s))
                .publishOn(Schedulers.parallel())
                .subscribe(customer -> cOut("|||", customer));
    }

    @Test
    public void test1() {
        Flux.range(1, 10000)
            .publishOn(Schedulers.parallel())
            .subscribe(customer -> cOut("|||", customer));
        System.out.println(Thread.currentThread().getName());
    }
}
